package com.zjw.demo.vertx.starter.plugin.impl.event;

import com.zjw.demo.vertx.starter.annotation.NonNull;
import com.zjw.demo.vertx.starter.annotation.Nullable;
import com.zjw.demo.vertx.starter.annotation.core.Plugin;
import com.zjw.demo.vertx.starter.core.ioc.IOCBeanInfo;
import com.zjw.demo.vertx.starter.core.ioc.IOCObserver;
import com.zjw.demo.vertx.starter.core.ioc.IOCScope;
import com.zjw.demo.vertx.starter.plugin.IPlugin;
import com.zjw.demo.vertx.starter.utils.LogUtils;
import com.zjw.demo.vertx.starter.core.Reflector;
import com.zjw.demo.vertx.starter.utils.ThreadPool;
import io.vertx.core.Handler;
import io.vertx.core.Vertx;
import io.vertx.core.eventbus.EventBus;
import io.vertx.core.eventbus.Message;

import java.lang.reflect.Method;
import java.lang.reflect.Parameter;

@Plugin
public final class EventBusPlugin implements IPlugin {

  @NonNull
  private static EventBus globalEventBus;

  private static final String TAG = "EventBusProxy";

  public static void proxy(Vertx vertx) {
    LogUtils.debug(TAG, "wrap vertx");
    globalEventBus = vertx.eventBus();
  }

  public static void sendMessage(String address, @Nullable Object message) {
    globalEventBus.send(address, message);
  }

  public static void publishMessage(String address, @Nullable Object message) {
    globalEventBus.publish(address, message);
  }

  public static void handleMessage(String address, Handler<Message<Object>> handler) {
    globalEventBus.consumer(address, handler);
  }

  public static EventBus getGlobalEventBus() {
    return globalEventBus;
  }

  @Override
  public void onPlugin(Vertx vertx) {
    dispatchSubscribe(vertx);
  }

  public static void dispatchSubscribe(Vertx vertx) {
    proxy(vertx);

    IOCScope.globalScope.registerRefreshObserver(new IOCObserver() {
      @Override
      public void onRefresh(Class<?> clazz) {
        if (clazz.isAnnotationPresent(Subscribe.class)) {
          Object object = Reflector.instanceEmptyConstructor(clazz);
          if (object == null) {
            return;
          }
          LogUtils.debug(TAG, "start parse " + object.getClass().getName());
          parse(object);
          IOCScope.globalScope.putBean(object.getClass(), object);
          LogUtils.debug(TAG, "end parse " + object.getClass().getName());
        }
      }

      @Override
      public void onInject(IOCBeanInfo o) {

      }

      @Override
      public void onComplete() {

      }
    });
  }

  private static void parse(Object object) {
    Method[] declaredMethods = object.getClass().getDeclaredMethods();
    for (Method declaredMethod : declaredMethods) {
      Consumer consumer = declaredMethod.getAnnotation(Consumer.class);
      if (consumer == null) {
        continue;
      }

      Parameter[] parameters = declaredMethod.getParameters();
      if (parameters.length != 1 || !parameters[0].getType().equals(Message.class)) {
        throw new IllegalArgumentException("consumer params must be io.vertx.core.eventbus.Message");
      }

      String address = consumer.address();
      boolean async = consumer.async();
      globalEventBus.consumer(address, event -> {
        invokeMethod(object, declaredMethod, async, event);
      });
    }
  }

  private static void invokeMethod(Object target, Method method, boolean async, Message<?> message) {
    if (async) {
      ThreadPool.submit(() -> {
        try {
          method.invoke(target, message);
        } catch (Exception e) {
          LogUtils.error(TAG, e);
        }
      });
    } else {
      try {
        method.invoke(target, message);
      } catch (Exception e) {
        LogUtils.error(TAG, e);
      }
    }
  }
}
